package com.hudsonmq.spring;

import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.request.PushMessageReq;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.json.JSONUtil;
import com.hudson.core.util.CollectionUtil;
import com.hudson.core.util.StringUtil;
import com.hudsonmq.spring.client.BrokerClient;
import com.hudsonmq.spring.client.NameSrvClient;
import com.hudsonmq.spring.config.HudsonMQConfig;
import com.hudsonmq.spring.domain.consume.PushRequest;
import com.hudsonmq.spring.impl.PushMessageServiceOneThread;
import com.hudsonmq.spring.impl.consume.HudsonMQConsumer;
import com.hudsonmq.spring.impl.consume.HudsonMQHandler;
import com.hudsonmq.spring.impl.produce.helper.TopicQueueHelper;
import com.hudsonmq.spring.utils.NettyUtil;
import com.hudsonmq.spring.utils.ThreadPoolHelper;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/8/17 12:18
 **/
@Component
public class HudsonMQClientController implements CommandLineRunner {

    private static final Logger log = LoggerFactory.getLogger(HudsonMQClientController.class);

    @Resource
    private HudsonMQConfig hudsonMQConfig;

    @Resource
    private List<HudsonMQHandler> handlerList;

    @Autowired(required = false)
    private PushMessageServiceOneThread pushMessageServiceOneThread;
    /**
     * 存放channel对应的名称
     * key：channelId
     * value:名称
     */
    public static final Map<String, String> CHANNEL_NAME_MAP = new ConcurrentHashMap<>();

    // 存放namesrv客户端
    private static final Map<String, Channel> NAME_SRV_CHANNEL_CLIENT = new ConcurrentHashMap<>();

    // 初始化连接NameSrv后,获取Broker的地址,每个NameSrv都进行请求获取,全部获取后再进行连接Broker;
    public static CountDownLatch START_COUNT_DOWN_LATCH;

    // 存放Broker Channel
    public static ConcurrentHashMap<String, Channel> BROKER_CHANNEL_MAP = new ConcurrentHashMap<>();

    /**
     * 存放Broker地址
     * key:brokerName
     * value:broker address
     */
    public static Map<String, String> BROKER_ADDRESS_MAP = new ConcurrentHashMap<>();

    // 初始化连接NameSrv后,获取QueueMap
    public static CountDownLatch START_GET_QUEUE_MAP_COUNT_DOWN_LATCH;

    /**
     * 所有Broker中的Topic以及对应的队列数
     * key:channel.id
     * value:topic信息
     */
    public static Map<String, Map<String, Integer>> BROKER_TOPIC_QUEUQ_MAP = new ConcurrentHashMap<>();


    @Override
    public void run(String... args) throws Exception {
        long start = System.currentTimeMillis();
        log.info("Hudson-MQ启动");

        List<InetSocketAddress> nameSrvAddressList = new ArrayList<>();
        for (String nameSrvAddress : hudsonMQConfig.getNameSrv().getNameSrvAddress()) {
            String[] address = nameSrvAddress.split(":");
            nameSrvAddressList.add(new InetSocketAddress(address[0], Integer.parseInt(address[1])));
        }

        // 启动连接NameSrv
        NameSrvClient.start(nameSrvAddressList, NAME_SRV_CHANNEL_CLIENT);

        START_COUNT_DOWN_LATCH = new CountDownLatch(NAME_SRV_CHANNEL_CLIENT.size());
        // 获取Broker的地址
        NameSrvClient.sendBrokerAddress(NAME_SRV_CHANNEL_CLIENT);

        //等待获取Broker地址
        boolean await = START_COUNT_DOWN_LATCH.await(10, TimeUnit.SECONDS);
        if (!await) {
            if (BROKER_ADDRESS_MAP.isEmpty()) {
                log.error("未获取Broker地址,程序退出");
                // 后续进行资源释放
                System.exit(0);
            }
            log.warn("存在NameSrv未返回Broker地址");
        }

        //连接Broker
        for (Map.Entry<String, String> brokerEntry : BROKER_ADDRESS_MAP.entrySet()) {
            String[] split = brokerEntry.getValue().split(":");
            ChannelFuture channelFuture = BrokerClient.getChannel(brokerEntry.getKey(), split[0], Integer.parseInt(split[1]));
            Channel channel;
            try {
                channel = channelFuture.sync().channel();
            } catch (InterruptedException e) {
                log.error("连接Broker失败：{}:{}", split[0], split[1]);
                throw new RuntimeException(e);
            }
            if (channel != null) {
                BROKER_CHANNEL_MAP.put(brokerEntry.getKey(), channel);
            }
        }

        START_GET_QUEUE_MAP_COUNT_DOWN_LATCH = new CountDownLatch(BROKER_CHANNEL_MAP.size());

        // 定时获取Broker队列
        ThreadPoolHelper.DEFAULT_SCHEDULED_POOL.scheduleAtFixedRate(() -> {
            for (Map.Entry<String, Channel> channelEntry : BROKER_CHANNEL_MAP.entrySet()) {
                BrokerClient.getTopicQueueInfo(channelEntry.getValue());
            }
        }, 0, 20, TimeUnit.SECONDS);

        //等待获取Broker queue_map
        boolean queueMapCountDown = START_GET_QUEUE_MAP_COUNT_DOWN_LATCH.await(10, TimeUnit.SECONDS);

        if (!queueMapCountDown) {
            if (TopicQueueHelper.TOPIC_QUEUE_MAP.isEmpty()) {
                log.error("获取队列失败,程序退出");
                // 后续进行资源释放
                System.exit(0);
            }
            log.warn("存在Broker未返回队列信息");
        }

        if(StringUtil.noEmpty(hudsonMQConfig.getConsumeGroup())){
            // 启动拉取消息线程
            pushMessageServiceOneThread.start();

            for (Map.Entry<String, Channel> entry : BROKER_CHANNEL_MAP.entrySet()) {
                startConsumer(entry.getValue());
            }
        }

        double l = (double) (System.currentTimeMillis() - start) / 1000;
        log.info("Hudson-MQ启动耗时:{}s", l);


    }


    public void startConsumer(Channel channel) {
        // 获取到 topic 路由信息后，开始初始化发送消息

        if (CollectionUtil.isEmpty(BROKER_TOPIC_QUEUQ_MAP.get(channel.id().asLongText()))) {
            return;
        }
        Map<String, Integer> queueMap = BROKER_TOPIC_QUEUQ_MAP.get(channel.id().asLongText());
        for (HudsonMQHandler hudsonMQHandler : handlerList) {
            HudsonMQConsumer annotation = hudsonMQHandler.getClass().getAnnotation(HudsonMQConsumer.class);
            if (annotation != null) {
                Integer i = queueMap.get(annotation.topic());
                for (int j = 0; j < i; j++) {
                    getPushMessage(channel, new PushRequest(channel, annotation.topic(), j, 32)).addListener(f -> {

                    });
                }
            }
        }
    }

    public ChannelFuture getPushMessage(Channel channel, PushRequest pushRequest) {
        PushMessageReq pushMessageReq = new PushMessageReq(hudsonMQConfig.getConsumeGroup(), pushRequest.getTopic(), pushRequest.getQueueId(), pushRequest.getSize());
        return NettyUtil.write(channel, RemotingCommand.builder()
                .type(MessageTypeEnum.PUSH_MESSAGE_REQ.getType())
                .body(JSONUtil.toJson(pushMessageReq).getBytes(StandardCharsets.UTF_8))
                .build());

    }


    /**
     * 判断是否可以继续拉取信息
     * @param pushRequest 拉取元数据
     * @return
     */
    public static boolean containsTopic(PushRequest pushRequest){
        Map<String, Integer> topicMap = BROKER_TOPIC_QUEUQ_MAP.get(pushRequest.getChannel().id().asLongText());
        if(CollectionUtil.isEmpty(topicMap)){
            return false;
        }
        Integer queueNum = topicMap.get(pushRequest.getTopic());
        if(pushRequest.getQueueId() > queueNum){
            return false;
        }
        return true;
    }

    /**
     * 断开连接后删除对应的数据
     * @param channelId channelId
     */
    public static void removeChannel(String channelId){
        String name = CHANNEL_NAME_MAP.remove(channelId);
        if(StringUtil.isEmpty(name)){
            return;
        }
        NAME_SRV_CHANNEL_CLIENT.remove(name);
        BROKER_CHANNEL_MAP.remove(name);
        BROKER_ADDRESS_MAP.remove(channelId);
        BROKER_TOPIC_QUEUQ_MAP.remove(channelId);
    }
}
